package com.iflytek.controller;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class _01OverSpeedControllerSQL {


    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        // 假如向kafka发送消息，这个表中就会有数据
        tenv.executeSql("CREATE TABLE table1 (\n" +
                "  `actionTime` BIGINT,\n" +
                "  `monitorId`  STRING,\n" +
                "  `cameraId`   STRING,\n" +
                "  `car`        STRING,\n" +
                "  `speed`      double,\n" +
                "  `roadId`     STRING,\n" +
                "  `areaId`     STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'topic-car',\n" +
                "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "  'properties.group.id' = 'c1',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");

        //2.声明mysql中的t_monitor_info对应的flinksql的表    维表
        tenv.executeSql("CREATE TABLE table2 (\n" +
                "  monitor_id    STRING,\n" +
                "  road_id       STRING,\n" +
                "  speed_limit   INT,\n" +
                "  area_id       STRING,\n" +
                "  PRIMARY KEY (monitor_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://hadoop102:3306/flink_project?useSSL=false&useUnicode=true&characterEncoding=utf8',\n" +
                "   'table-name' = 't_monitor_info',\n" +
                "   'username' = 'root',\n" +
                "   'password' = '12345678'\n" +
                ")");

        //4.查询
        tenv.executeSql("create view table4 as select car,monitorId,roadId,speed,speed_limit,actionTime " +
                "from( " +
                "   select t1.*,if(t2.monitor_id is null,60,t2.speed_limit) speed_limit " +
                "   from table1 t1 left join table2 t2 " +
                "   on t1.monitorId = t2.monitor_id " +
                ")t3 where speed > speed_limit * 1.1");

        //3.声明mysql中的t_speeding_info对应的flinksql中的表       sink
        tenv.executeSql("CREATE TABLE table3 (\n" +
                "  car           STRING,\n" +
                "  monitor_id    STRING,\n" +
                "  road_id       STRING,\n" +
                "  real_speed    DOUBLE,\n" +
                "  limit_speed   INT,\n" +
                "  action_time   BIGINT,\n" +
                "  PRIMARY KEY (car,action_time) NOT ENFORCED" +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://hadoop102:3306/flink_project?useSSL=false&useUnicode=true&characterEncoding=utf8',\n" +
                "   'table-name' = 't_speeding_info',\n" +
                "   'username' = 'root',\n" +
                "   'password' = '12345678'\n" +
                ")");

        //5.写入
        tenv.executeSql("insert into table3 select * from table4");


    }
}
